package com.hunliji.monitor.share.component.mq.rocket.impl;

import com.alibaba.fastjson.JSON;
import com.hunliji.common.mq.MsgData;
import com.hunliji.monitor.share.component.mq.rocket.RocketMqFacade;
import com.hunliji.monitor.share.component.mq.rocket.constants.RocketMqConstants;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * rocket mq 消息发送器
 * @author wuchen
 */
@Component
public class RocketMqFacadeImpl implements RocketMqFacade {

    private static final Logger logger = LoggerFactory.getLogger(RocketMqFacadeImpl.class);
    @Resource
    private RocketMQTemplate rocketMqTemplate;



    /**
     * 发送消息 - 同步
     * 超时时间 {@link RocketMqConstants#DEFAULT_TIME_OUT}
     * @param topic topic
     * @param tag   tag
     * @param msg  消息内容
     */
    @Override
    public void send(String topic, String tag, Object msg) {
        send(topic,tag,msg,RocketMqConstants.DEFAULT_TIME_OUT);
    }

    /**
     * 发送消息 - 同步
     * @param topic topic
     * @param tag   tag
     * @param msg  消息内容
     * @param timeout 发送消息的超时时间
     */
    @Override
    public void send(String topic, String tag, Object msg, Long timeout) {
        MsgData msgData = new MsgData<>(RocketMqConstants.SEND_APP_NAME,msg);
        SendResult sendResult = rocketMqTemplate.syncSend(topic + ":" + tag, MessageBuilder.withPayload(msgData).build(),timeout);
        logger.info("business 发送消息-(topic:{},tag:{},msg:{},result:{})",
                topic,tag,JSON.toJSONString(msg), sendResult);
    }

    /**
     * 发送延时消息
     * @param topic topic
     * @param tag tag
     * @param msg 消息内容
     * @param delayLevel 延时多久
     */
    @Override
    public void delayedSend(String topic, String tag, Object msg, RocketMqConstants.DelayLevel delayLevel) {
        delayedSend(topic,tag,msg,RocketMqConstants.DEFAULT_TIME_OUT,delayLevel);
    }

    /**
     * 发送延时消息
     * @param topic topic
     * @param tag tag
     * @param msg 消息内容
     * @param timeout 发送消息的超时时间
     * @param delayLevel 延时多久
     */
    @Override
    public void delayedSend(String topic, String tag, Object msg, Long timeout, RocketMqConstants.DelayLevel delayLevel) {
        MsgData msgData = new MsgData<>(RocketMqConstants.SEND_APP_NAME,msg);
        SendResult sendResult = rocketMqTemplate.syncSend(topic + ":" + tag, MessageBuilder.withPayload(msgData).build(),timeout,delayLevel.getLevel());
        logger.info("business 发送延时消息 topic:{},tag:{},delayLevel:{},msg:{},result:{}",
                topic,tag,delayLevel.name(),JSON.toJSONString(msg), sendResult);
    }

    /**
     * 顺序消费
     * 在这种模式下，会根据 orderKey 的 hash去选择发送到哪个队列
     * 具体可看{@link SelectMessageQueueByHash}
     * 发送顺序消息，我只能保障我hash相同的消息发送到相同的队列
     * 如果需要顺序消费，还需要在消费者中做一定的限制
     * 这里暂时没有找到设置超时时间的地方，默认的超时时间 = 3000ms 具体见{@link DefaultMQProducer#sendMsgTimeout}
     * @param topic topic
     * @param tag tag
     * @param msg 消息内容
     * @param orderKey：发送顺序消息的标识符
     */
    @Override
    public void orderSend(String topic, String tag, Object msg, String orderKey) {
        MsgData msgData = new MsgData<>(RocketMqConstants.SEND_APP_NAME,msg);
        rocketMqTemplate.sendOneWayOrderly(topic + ":" + tag,MessageBuilder.withPayload(msgData).build(),orderKey);
        logger.info("business 发送延时消息 topic:{},tag:{},orderKey:{},msg:{}", topic,tag,orderKey,JSON.toJSONString(msg));
    }


}
